package com.itcj.dmp.tags

import com.itcj.dmp.casename.tagcase
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object HistoryTags {

  def getHistoryTag(sparkSession: SparkSession): Option[Dataset[tagcase]] = {
    import com.itcj.dmp.utils.KuduHelper._
    import com.itcj.dmp.tags.TagRunner._
    import com.itcj.dmp.casename.TableName._
    //获取history的tag表，如果有，进行加权削减，如果没有返回None
    val mf = sparkSession.readKuduTable(yesterdayTagsTableName)
    if (mf.isDefined) {
      val hsds: DataFrame = mf.get
      import sparkSession.implicits._
      val hsds1: Dataset[tagcase] =hsds.map(row=>{
        val tags: Map[String, Double] = row.getAs[Map[String,Double]]("tags")
        val tags1 = tags.map(kv=>(kv._1,kv._2*coefficient))
        tagcase(row.getAs[String]("mainIp"),row.getAs[Map[String,String]]("ips"),tags1)
      })
      Some(hsds1)
    }else{None}
  }


}
